package mqs

import (
	"context"
	"encoding/json"
	"fmt"
	"property_service/internal/svc"
	"property_service/mqs/message"
	"sync"

	"github.com/zeromicro/go-queue/kq"
	"github.com/zeromicro/go-zero/core/logx"
)

type ActionRun struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	pusher chan string
}

var (
	actionRun *ActionRun
	lock      sync.Mutex
)

func NewActionRun(ctx context.Context, svcCtx *svc.ServiceContext, producer *kq.Pusher) *ActionRun {
	lock.Lock()
	defer lock.Unlock()

	if actionRun == nil || actionRun.pusher == nil {
		if ctx == nil {
			ctx = context.Background()
		}
		actionRun = &ActionRun{
			ctx:    ctx,
			svcCtx: svcCtx,
			pusher: make(chan string),
		}
		go actionRun.asyncPush(ctx, producer, actionRun.pusher)
	}
	return actionRun

}

func (l *ActionRun) asyncPush(ctx context.Context, producer *kq.Pusher, pusher <-chan string) {
	var msg message.ClientMsg
	for data := range pusher {
		fmt.Printf("starting write msg %v", data)
		e := json.Unmarshal([]byte(data), &msg)
		if e != nil {
			return
		}
		switch msg.Code {
		case message.AsyncComm:
			if err := producer.Push(ctx, data); err != nil {
				fmt.Printf("async wirte error: %s \n", err)
				return
			}
		default:
			logx.Errorf("msg.Code err")
		}
	}
	return
}

func Push(ctx *svc.ServiceContext, data string) {
	if actionRun == nil {
		actionRun = NewActionRun(ctx.Ctx, ctx, CreateKafkaProducer(&ctx.Config))
	}
	actionRun.pusher <- data
}

func (l *ActionRun) Close() {
	close(l.pusher)
}
